/*
 * Copyright 2015 MingxingWang

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 *     http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.graphx

import java.io.{DataInputStream, DataOutputStream}

import com.audaque.graphdb._
import com.audaque.graphdb.util.GraphDBUtil
import org.apache.spark.graphx.graphdb.MyVertexPartition._
import org.apache.spark.graphx.impl.VertexPartition

import scala.collection.mutable

/**
 * Created by Ming on 2015/3/26.
 */
class LVertices(var partitions: List[VertexPartition[Property]] = List[VertexPartition[Property]](),
                val addBuffer: mutable.Map[VertexId, Property] = new mutable.HashMap(),
                var addBuffSize: Int = 0) {

  def count(): Long = {
    val pSize = {
      val size = partitions.map(_.size)
      if (size.isEmpty) 0 else size.reduce(_ + _)
    }
    val total = pSize + addBuffSize
    println("Total:" + total + ", partition:" + pSize + ", add buffer:" + addBuffer.size)
    total
  }

  def find(vid: VertexId): Option[Vertex] = {
    addBuffer.get(vid) match {
      case Some(x) => return Some(Vertex(vid, x))
      case None =>
    }
    for (p <- partitions) {
      if (p.isDefined(vid)) return Some(Vertex(vid, p(vid)))
    }
    None
  }

  def contain(vid: VertexId): Boolean = {
    if (addBuffer.contains(vid)) return true
    for (p <- partitions) {
      if (p.isDefined(vid)) {
        return true
      }
    }
    false
  }

  def delete(vid: VertexId): Boolean = {
    if (addBuffer.contains(vid)) {
      addBuffer.remove(vid)
      return true
    }
    for (p <- partitions) {
      if (p.delete(vid)) {
        return true
      }
    }
    false
  }

  /**
   * 增加顶点
   * @param vid id
   * @param prop 属性
   * @param replace 顶点冲突时是否替换原值
   * @return 是否成功
   */
  def add(vid: VertexId, prop: Property, replace: Boolean = false): Boolean = {
    if (addBuffer.contains(vid)) {
      if (replace) {
        addBuffer += (vid -> prop)
        return true
      } else {
        return false
      }
    }
    if (replace) {
      // 检查partitions中是否存在，存在则替换
      for (p <- partitions) {
        if (p.update(vid, prop)) {
          return true
        }
      }
    } else {
      // 检查partitions中是否存在，存在则false
      for (p <- partitions) {
        if (p.isDefined(vid)) {
          return false
        }
      }
    }
    addBuffSize += 1
    addBuffer += (vid -> prop)
    if (addBuffSize >= GraphDBConf.VertexBufferMaxSize)
      update()
    true
  }

  /**
   * addbuffer存在则直接更新，否则partition中被删除则返回false。
   */
  def modify(vid: VertexId, prop: Property): Boolean = {
    if (addBuffer.contains(vid)) {
      addBuffer += (vid -> prop)
      return true
    }
    for (p <- partitions) {
      if (p.update(vid, prop)) {
        return true
      }
    }
    false
  }

  def update(): Unit = {
    partitions = partitions :+ VertexPartition(addBuffer.toIterator)
    addBuffer.clear()
    addBuffSize = 0
  }

  def save(out: DataOutputStream) = {
    partitions.foreach(p => p.iterator.foreach(v => GraphDBUtil.saveVertex(Vertex(v._1, v._2), out)))
    addBuffer.foreach(v => GraphDBUtil.saveVertex(Vertex(v._1, v._2), out))
  }

  def load(in: DataInputStream, schFact: SchemaFactory) = {
    partitions = List[VertexPartition[Property]]()
    var builder = new mutable.HashMap[VertexId, Property]()
    var v = GraphDBUtil.loadVertex(in, schFact)
    while (v != null) {
      builder += (v.vid -> v.prop)
      if (builder.size >= GraphDBConf.VertexBufferMaxSize) {
        partitions = partitions :+ VertexPartition(builder.toIterator)
        builder.clear()
      }
      v = GraphDBUtil.loadVertex(in, schFact)
    }
    if (builder.size > 0) {
      partitions = partitions :+ VertexPartition(builder.toIterator)
      builder = null
    }
  }

  def clear() = {
    addBuffer.clear()
    addBuffSize = 0
    partitions = List[VertexPartition[Property]]()
  }
}
